package com.ibm.cps.redis;

import java.util.ArrayList;
import java.util.List;


import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Pipeline;
import storm.trident.state.map.IBackingMap;

public class RuleStateDB<T> implements IBackingMap<T> {

	private JedisPool pool;
	private KeyBuilder keyBuilder;

	public RuleStateDB(String redisHost, int port) {
		pool = new JedisPool(new JedisPoolConfig(), redisHost, port);
		keyBuilder = new DefaultKeyBuilder();
	}

	@Override
	public List<T> multiGet(List<List<Object>> keys) {
		Jedis jedis = pool.getResource();
		String[] keysList = keyBuilder.build(keys);
		List<String> valuesList = jedis.mget(keysList);
//		for (int i = 0; i < keysList.length; i++) {
//			System.out.println("key " + keysList[i] + "value "
//					+ valuesList.get(i));
//		}
		return deserializeValues(keys.size(), valuesList);
	}

	private List<T> deserializeValues(int keySize, List<String> values) {
		List<T> result = new ArrayList<T>(keySize);
		for (String value : values) {
			result.add((T) value);
		}
		return result;
	}

	@Override
	public void multiPut(List<List<Object>> keys, List<T> vals) {
		if (keys.size() == 0) {
			return;
		}

		Jedis jedis = pool.getResource();
		try {
			Pipeline pipeline = jedis.pipelined();
			String[] newKeys = keyBuilder.build(keys);
			String[] keyValues = constructKeyValues(newKeys, vals);
			pipeline.mset(keyValues);
			pipeline.multi();
			pipeline.exec();
			pipeline.sync();
		} finally {
			pool.returnResource(jedis);
		}
	}

	private String[] constructKeyValues(String[] newKeys, List<T> vals) {
		if (newKeys == null || newKeys.length <= 0) {
			return null;
		}
		if (newKeys.length != vals.size()) {
			System.err.println("Keys and values should be the same size.");
			System.exit(-1);
		}
		String[] keyValues = new String[newKeys.length * 2];
		for (int i = 0; i < newKeys.length; i++) {
			keyValues[2 * i] = newKeys[i];
			keyValues[2 * i + 1] = vals.get(i).toString();
		}
		return keyValues;
	}
}
